package cn.ymatrix.connectors;

import cn.ymatrix.utils.BatchProcessingUtils;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

public class MySqlToYMatrix implements DebeziumDeserializationSchema<String> {

    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        Struct valueStruct = (Struct) sourceRecord.value();
        Struct sourceStrut = valueStruct.getStruct("source");
        String database = sourceStrut.getString("db");
        String table = sourceStrut.getString("table");

        String type = Envelope.operationFor(sourceRecord).toString().toLowerCase();
        if (type.equals("create")) {
            type = "insert";
        }
        JSONObject jsonObj = new JSONObject();
        jsonObj.put("database", database);
        jsonObj.put("table", table);
        jsonObj.put("type", type);

        //获取数据data,delete获取的是before的数据
        Struct afterStruct = null;
        if (type.equals("delete")) {
            afterStruct = valueStruct.getStruct("before");
        } else {
            afterStruct = valueStruct.getStruct("after");
        }

        JSONObject dataJsonObj = new JSONObject();
        if (afterStruct != null) {
            for (Field field : afterStruct.schema().fields()) {
                String fieldName = field.name();
                Object fieldValue = afterStruct.get(field);
                dataJsonObj.put(fieldName, fieldValue);
            }
        }
        jsonObj.put("cdcdata", dataJsonObj);
        // 向下传递数据
        BatchProcessingUtils.sqlTypeBatchSplicing(jsonObj, 500);
        // 向下游传递数据
        // collector.collect(jsonObj.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return TypeInformation.of(String.class);
    }

}
